package com.hcy.orderapi.mqconsumer;

import com.alibaba.fastjson.JSONObject;
import com.hcy.orderbusiness.Impl.OptOrderBzImpl;
import com.hcy.rabbitbasicmodel.common.MyWebSocketMessage;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Service;

import java.util.Arrays;

/**
 * @Author huchenying
 * @Description
 * @Date 2022/3/19 21:47
 */
@Service
public class MQClearTimeoutCallLiveConsumer implements ApplicationRunner {

    private static final Logger logger = LoggerFactory.getLogger(MQClearTimeoutCallLiveConsumer.class);

    @Value("${MQ.namesrvAddr}")
    private String addr;

    @Autowired
    OptOrderBzImpl optOrderBz;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("DelayCallLive");
        //设置NameServer地址
        consumer.setNamesrvAddr(addr);
        //设置起始消费位置
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //指定消费Topic和Tag
        consumer.subscribe("ClearTimeoutCall","*");
        //指定消费模式 默认集群模式
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.setConsumeThreadMin(1);
        consumer.setConsumeThreadMax(1);
        consumer.setConsumeMessageBatchMaxSize(8);
        //注册消息监听器
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            //逐条消费消息
            for (MessageExt msg : msgs) {
                //订单 id
                String msgId = msg.getKeys();
                //取消订单
                consumer(msgId);
            }
            //返回消费状态
            //重新消费
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        //开启消费
        consumer.start();
        logger.info("websocket mq consumer started");
    }

    public void consumer(String orderNumber){
        logger.info("consumer start param -> orderNumber = {}", orderNumber);
        try {
            optOrderBz.autoCancelOrder(orderNumber);
        } catch (Exception e) {
            logger.error("cancel order fail orderNumber = [{}]",ToStringBuilder.reflectionToString(orderNumber));
        }
    }
}
